package com.wmh.baseservice.common.redis.mq.core;


import com.alibaba.fastjson2.JSONObject;
import com.wmh.baseservice.common.redis.mq.core.interceptor.RedisMessageInterceptorFactory;
import com.wmh.baseservice.common.redis.mq.core.stream.AbstractRedisStreamMessage;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.connection.stream.StreamInfo;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.core.StringRedisTemplate;

import java.util.HashSet;
import java.util.Objects;
import java.util.Set;

/**
 * Redis MQ 操作模版类
 *
 * @author 审心
 */
@Getter
@Slf4j
@RequiredArgsConstructor
public class RedisMqTemplate {

    /**
     * 健前缀
     */
    public static final String MQ_STREAM_KEY_PREFIX = "stream-mq:";

    /**
     * 项目所有 stream key
     */
    public static final Set<String> STREAM_KEY_LIST = new HashSet<>();

    private final StringRedisTemplate stringRedisTemplate;

    private final RedisMessageInterceptorFactory redisMessageInterceptorFactory;


    /**
     * 发送 stream 消息
     *
     * @param message 消息体
     */
    public RecordId send(AbstractRedisStreamMessage message) {
        try {
            redisMessageInterceptorFactory.sendBefore(message);
            StreamOperations<String, Object, Object> ops = stringRedisTemplate.opsForStream();
            this.streamTrim(message.getStreamKey(), message.getMaxSize());
            // 发布消息
            return ops.add(StreamRecords.newRecord()
                    .ofObject(JSONObject.toJSONString(message))
                    .withStreamKey(message.getStreamKey()));
        } finally {
            redisMessageInterceptorFactory.sendAfter(message);
        }
    }

    /**
     * 当 stream 长度 + 1 达到 maxLength 值时 裁剪流
     *
     * @param key       健
     * @param maxLength 最大长度
     */
    public void streamTrim(String key, Long maxLength) {
        StreamOperations<String, Object, Object> ops = stringRedisTemplate.opsForStream();
        // 限制队列长度
        Long size = ops.size(key);
        if (size != null && size + 1 >= maxLength) {
            ops.trim(key, maxLength);
        }
    }

    /**
     * 创建一个消费者组
     *
     * @param template 操作模版
     * @param key      键值
     * @param group    组名
     */
    public static void createGroup(StringRedisTemplate template, String key, String group) {
        // key 不存在时
        if (Boolean.FALSE.equals(template.hasKey(key))) {
            template.opsForStream().createGroup(key, group);
            return;
        }
        // 组 不存在时
        StreamInfo.XInfoGroups groups = template.opsForStream().groups(key);
        if (groups.stream().noneMatch(g -> Objects.equals(group, g.groupName()))) {
            template.opsForStream().createGroup(key, group);
        }
    }
}
